package com.examples.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.BufferedOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class BaiduHotToper {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<JSONArray> dfdgdg = env.addSource(new RichSourceFunction<JSONArray>() {
            private static final long serialVersionUID = 33342353984591675L;


            @Override
            public void run(SourceContext<JSONArray> sourceContext) throws Exception {
                OkHttpClient client = new OkHttpClient().newBuilder()
                        .build();
                Request request = new Request.Builder()
                        .url("https://top.baidu.com/board?tab=realtime")
                        .method("GET", null)
                        .addHeader("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0")
                        .build();
                Response response = client.newCall(request).execute();
                String string = response.body().string();

                Pattern regex = Pattern.compile("\\<\\!\\-\\-s\\-data\\:(.*?)\\-\\-\\>");
                Matcher matcher = regex.matcher(string);
                while (matcher.find()){
                    try {
                        String group = matcher.group(1);
                        JSONObject t = JSON.parseObject(group);
                        JSONObject data = (JSONObject) t.get("data");
                        JSONArray jsonArray = (JSONArray) data.get("cards");
                        JSONObject jsonObject = (JSONObject) jsonArray.get(0);
                        JSONArray jsonArray1 = (JSONArray) jsonObject.get("content");
                        sourceContext.collect(jsonArray1);
                    }catch (Exception ignored){}
                }

            }

            @Override
            public void cancel() {
            }
        }).setParallelism(1).name("baiduHotTopSource");
        SingleOutputStreamOperator<JSONObject> baiduHotTopFlatMap = dfdgdg.flatMap(new FlatMapFunction<JSONArray, JSONObject>() {
            @Override
            public void flatMap(JSONArray objects, Collector<JSONObject> collector) throws Exception {
                objects.forEach(o -> {
                    collector.collect((JSONObject) o);
                });
            }
        }).setParallelism(5).name("baiduHotTopFlatMap");
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> tuple2SingleOutputStreamOperator = baiduHotTopFlatMap.flatMap(new FlatMapFunction<JSONObject, Tuple2<String, JSONObject>>() {
            @Override
            public void flatMap(JSONObject jsonObject, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                collector.collect(new Tuple2<String, JSONObject>(String.valueOf(System.currentTimeMillis()), jsonObject));
            }
        });
        OutputTag<Tuple2<String, String>> imageUrl = new OutputTag<Tuple2<String, String>>("imageUrl", Types.TUPLE(Types.STRING, Types.STRING));
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> singleOutputStreamOperator = tuple2SingleOutputStreamOperator.process(new ProcessFunction<Tuple2<String, JSONObject>, Tuple2<String, JSONObject>>() {
            @Override
            public void processElement(Tuple2<String, JSONObject> stringJSONObjectTuple2, Context context, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                collector.collect(stringJSONObjectTuple2);
                context.output(imageUrl, new Tuple2<String, String>(String.valueOf(System.currentTimeMillis()), stringJSONObjectTuple2.f1.getString("img")));
            }
        });
        SideOutputDataStream<Tuple2<String, String>> sideImageUrlOutput = singleOutputStreamOperator.getSideOutput(imageUrl);
        SingleOutputStreamOperator<Tuple2<String, String>> tuple2SingleOutputStreamOperator1 = sideImageUrlOutput.flatMap(new FlatMapFunction<Tuple2<String, String>, Tuple2<String, String>>() {
            @Override
            public void flatMap(Tuple2<String, String> stringStringTuple2, Collector<Tuple2<String, String>> collector) throws Exception {
                OkHttpClient client = new OkHttpClient();
                Request request = new Request.Builder()
                        .url(stringStringTuple2.f1)
                        .build();

                try (Response response = client.newCall(request).execute()) {
                    if (!response.isSuccessful()) {
                        throw new IOException("Unexpected code " + response);
                    }

                    ResponseBody body = response.body();
                    if (body != null) {
                        // 一次性读取所有字节
                        byte[] imageBytes = body.bytes();
                        String name = "D:\\ffffff\\2024-01-29--15\\" + stringStringTuple2.f0 + ".jpg";
                        BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(name));
                        out.write(imageBytes);
                        out.close();
                        stringStringTuple2.f1 = name;
                        collector.collect(stringStringTuple2);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        // 将侧枝流合并到主支： 将 tuple2SingleOutputStreamOperator1 合并到 singleOutputStreamOperator
        ConnectedStreams<Tuple2<String, JSONObject>, Tuple2<String, String>> connect = singleOutputStreamOperator.connect(tuple2SingleOutputStreamOperator1);
        ConnectedStreams<Tuple2<String, JSONObject>, Tuple2<String, String>> tuple2Tuple2ConnectedStreams = connect.keyBy(s1 -> s1.f0, s2 -> s2.f0);
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> imgLocalFile = tuple2Tuple2ConnectedStreams.process(new CoProcessFunction<Tuple2<String, JSONObject>, Tuple2<String, String>, Tuple2<String, JSONObject>>() {
            private transient MapStateDescriptor<String, List<Tuple2<String, String>>> cacheDescriptor;
            private transient ValueStateDescriptor<Tuple2<String, JSONObject>> pendingElements;

            @Override
            public void open(Configuration parameters) throws Exception {
                cacheDescriptor = new MapStateDescriptor<>("cache", BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint<List<Tuple2<String, String>>>() {}));
                pendingElements = new ValueStateDescriptor<>("pendingElements", TypeInformation.of(new TypeHint<Tuple2<String, JSONObject>>() {}));

                // 初始化缓存状态
                getRuntimeContext().getMapState(cacheDescriptor);
                getRuntimeContext().getState(pendingElements);
                super.open(parameters);
            }

            @Override
            public void processElement1(Tuple2<String, JSONObject> stringJSONObjectTuple2, Context context, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                if (getRuntimeContext().getMapState(cacheDescriptor).contains(stringJSONObjectTuple2.f0)) {
                    List<Tuple2<String, String>> tuple2s = getRuntimeContext().getMapState(cacheDescriptor).get(stringJSONObjectTuple2.f0);
                    for (Tuple2<String, String> s2Element : tuple2s) {
                        if(s2Element.f0.equals(stringJSONObjectTuple2.f0)){
                            stringJSONObjectTuple2.f1.put("imgLocalFile", s2Element.f1);
                            collector.collect(stringJSONObjectTuple2);
                        }
                    }
                } else {
                    // 缓存中没有对应数据时暂存此元素，等待第二个流的数据到来
                    getRuntimeContext().getState(pendingElements).update(stringJSONObjectTuple2);
                }
            }

            @Override
            public void processElement2(Tuple2<String, String> stringStringTuple2, Context context, Collector<Tuple2<String, JSONObject>> collector) throws Exception {
                // 当收到第二个流的数据时，检查是否有待处理的第一个流的数据
                Tuple2<String, JSONObject> pending = getRuntimeContext().getState(pendingElements).value();
                if (pending != null && pending.f0.equals(stringStringTuple2.f0)) {
                    // 将第二个流的数据添加到缓存，并处理先前暂存的第一个流的数据
                    List<Tuple2<String, String>> list = getRuntimeContext().getMapState(cacheDescriptor).get(stringStringTuple2.f0);
                    if (list == null) {
                        list = new ArrayList<>();
                    }
                    list.add(stringStringTuple2);
                    getRuntimeContext().getMapState(cacheDescriptor).put(stringStringTuple2.f0, list);

                    // 更新并输出第一个流的数据
                    pending.f1.put("imgLocalFile", stringStringTuple2.f1);
                    collector.collect(null);
                    getRuntimeContext().getState(pendingElements).clear();
                } else {
                    // 如果没有匹配的元素，直接将新数据加入缓存
                    List<Tuple2<String, String>> list = getRuntimeContext().getMapState(cacheDescriptor).get(stringStringTuple2.f0);
                    if (list == null) {
                        list = new ArrayList<>();
                    }
                    list.add(stringStringTuple2);
                    getRuntimeContext().getMapState(cacheDescriptor).put(stringStringTuple2.f0, list);
                }
            }
        });
        SingleOutputStreamOperator<Tuple2<String, JSONObject>> filter = imgLocalFile.filter(new FilterFunction<Tuple2<String, JSONObject>>() {
            @Override
            public boolean filter(Tuple2<String, JSONObject> stringJSONObjectTuple2) throws Exception {
                if (stringJSONObjectTuple2 == null) {
                    return false;
                }
                return true;
            }
        });
        filter.addSink(new SinkFunction<Tuple2<String, JSONObject>>() {
            @Override
            public void invoke(Tuple2<String, JSONObject> value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                System.out.println("value = " + value);
            }
        });


        env.execute("BaiduHotToper");


    }
}
